package fox.framework.rpc.client.node;

import fox.framework.rpc.client.node.rule.Rule;
import fox.framework.rpc.converter.RpcMessageConverter;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;

/**
 * @author cuichao
 * @Description: RootRpcMetadata
 * @Date: create in 2021/1/11 15:55
 */
public class RootRpcMetadata implements RpcMetadata {

    private String metaspace;

    private Rule rule;

    private boolean lazy;

    private volatile boolean fetched;

    private List<RpcNode> allNodes = new ArrayList<>();

    private List<RpcNode> availableNodes = new ArrayList<>();

    private List<RpcNode> pauseNodes = new ArrayList<>();

    private RpcMessageConverter messageConverter;



    /**
     * 更新nodes锁
     */
    private Lock nodeLock = new ReentrantLock();
    /**
     * 获取meta node列表锁(Rpc-Client 执行请求时需要远程获取节点列表时用)
     */
    private Lock metaLock = new ReentrantLock();

    public RootRpcMetadata(String metaspace) {
        this.metaspace = metaspace;
    }




    public void metaLock(){
        metaLock.lock();
    }

    public void releaseLock(){
       metaLock.unlock();
    }

    @Override
    public void setNodeState(RpcNode node, int state) {
        nodeLock.lock();
        try {

            switch (state) {
                case RpcNode.RPC_NODE_READY:
                case RpcNode.RPC_NODE_RUNNING:
                    if (!availableNodes.contains(node)) {
                        availableNodes = addNode(availableNodes, node);
                        pauseNodes = removeNode(pauseNodes, node);
                    }
                    break;
                case RpcNode.RPC_NODE_PAUSE:
                    if (!pauseNodes.contains(node)) {
                        pauseNodes = addNode(pauseNodes, node);
                        availableNodes = removeNode(availableNodes, node);
                    }
                    break;
                case RpcNode.RPC_NODE_SHUTDOWN:
                default:
                    //nothing to do
            }
            node.setState(state);
        } finally {
            nodeLock.unlock();
        }
    }

    @Override
    public String metaspace() {
        return metaspace;
    }

    @Override
    public boolean containNode(RpcNode node) {
        return allNodes.contains(node);
    }

    @Override
    public List<RpcNode> getAllNodes() {
        return Collections.unmodifiableList(allNodes);
    }

    @Override
    public List<RpcNode> getAvailableNodes() {
        nodeLock.lock();
        try {
            return Collections.unmodifiableList(availableNodes);
        } finally {
            nodeLock.unlock();
        }

    }

    @Override
    public List<RpcNode> getPauseStateNodes() {
        return Collections.unmodifiableList(pauseNodes);
    }


    @Override
    public void addNode(RpcNode... nodes) {
        nodeLock.lock();
        try {
            List<RpcNode> addNodes = new ArrayList<>();
            //过滤掉已经有的node
            for (RpcNode node : nodes) {
                if (!allNodes.contains(node)) {
                    addNodes.add(node);
                }
            }

            if (addNodes.isEmpty()) {
                return;
            }

            List<RpcNode> oldAllList = allNodes;
            List<RpcNode> newAllList = new ArrayList<>();
            newAllList.addAll(oldAllList);
            newAllList.addAll(addNodes);
            List<RpcNode> newAvailableNodes = new ArrayList<>();
            List<RpcNode> newPauseStateNodes = new ArrayList<>();
            newAllList.forEach(v -> {
                if(v == null){
                    return;
                }
                if (v.getState() >= 0) {
                    newAvailableNodes.add(v);
                } else {
                    newPauseStateNodes.add(v);
                }
            });
            allNodes = newAllList;
            availableNodes = newAvailableNodes;
            pauseNodes = newPauseStateNodes;
            //help GC
            oldAllList = null;
        } finally {
            nodeLock.unlock();
        }

    }

    @Override
    public Rule rule() {
        return rule;
    }

    @Override
    public RpcMessageConverter getMessageConverter() {
        return messageConverter;
    }

    @Override
    public boolean isLazy() {
        return lazy;
    }

    @Override
    public boolean isFetched() {
        return fetched;
    }


    private List<RpcNode> addNode(List<RpcNode> nodes, RpcNode addNode) {
        List<RpcNode> newList = new ArrayList<>();
        newList.addAll(nodes);
        newList.add(addNode);
        return newList;
    }


    private List<RpcNode> removeNode(List<RpcNode> nodes, RpcNode removeNode) {
        return nodes.stream().filter(v -> v != removeNode).collect(Collectors.toList());
    }


    public void setRule(Rule rule) {
        this.rule = rule;
    }

    public void setLazy(boolean lazy) {
        this.lazy = lazy;
    }

    public void setFetched(boolean fetched) {
        this.fetched = fetched;
    }

    public void setMessageConverter(RpcMessageConverter messageConverter) {
        this.messageConverter = messageConverter;
    }

}
